package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo6Map {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()

    conf.setAppName("map")

    conf.setMaster("local")

    val sc = new SparkContext(conf)

    /**
      * 构建qrdd方法
      * 1、读取文件
      * 2、基于scala集合构建rdd --- 一班用于测试
      *
      */

    //基于scala集合构建rdd
    val rdd1: RDD[Int] = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8, 9), 2)

    //getNumPartitions 不是一个算子，是一个普通的方法
    println(s"rdd1:${rdd1.getNumPartitions}")
    /**
      * map算子： 将rdd中的数据一行一行传递给后面的函数，及那个函数的返回值构建成一个新的rdd
      *
      */

    val rdd2: RDD[Int] = rdd1.map((i: Int) => i + 1)

    rdd2.foreach(println)


    /**
      * mapPartitions: 将一个分区的数据传递给后面的函数，
      * 一次处理一个分区的数据，需要返回一个迭代器
      *
      * 为什么是迭代器而不是集合，因为集合会将数据加载到内存中，
      * 如果一个分区数据量太大会导致内存溢出
      *
      */
    val rdd3: RDD[Int] = rdd1.mapPartitions((iter: Iterator[Int]) => {
      println("=" * 100)
      //对这个迭代器进行处理，这里是scala api
      val iterator: Iterator[Int] = iter.map((i: Int) => i * 2)
      //返回一个迭代器
      iterator
    })

    rdd3.foreach(println)

    /**
      * mapPartitionsWithIndex: 多了分区编号
      */

    val rdd4: RDD[Int] = rdd1.mapPartitionsWithIndex {
      case (index: Int, iter: Iterator[Int]) =>
        println(s"当前分区的编号:$index")

        iter
    }

    rdd4.foreach(println)


  }

}
